﻿#ifdef AS_REDIS
#include "as_redis_client.h"
#include "as_redis_exception.h"
#include "as_redis_result.h"
#include "rapidjson/document.h"
#include "rapidjson/writer.h"
#include "rapidjson/stringbuffer.h"
#include <chrono>
#include <fstream>

as::db::RedisConf::RedisConf(const std::string &path)
{
	std::ifstream fs(path);
	if (fs.is_open() == false)
		return;

	std::string tmp, context;
	while (getline(fs, tmp))
	{
		context.append(tmp);
		tmp.clear();
	}

	rapidjson::Document doc;
	doc.Parse(context.c_str());
	if (doc.HasParseError())
	{
		fs.close();
		assert(false);
	}

	host = doc["hostname"].GetString();
	pw = doc["pw"].GetString();
	port = doc["port"].SafeGetInt();
	timeout.tv_sec = doc["timeout"][0].SafeGetInt();
	timeout.tv_usec = doc["timeout"][1].SafeGetInt();
	fs.close();
}

as::db::RedisConf::RedisConf(const std::string h, const std::string password, int p, timeval tv)
	:host(h), pw(password), port(p), timeout(tv)
{

}

as::db::RedisClient::RedisClient(redisContext *con) : context_(con)
{

}

as::db::RedisClient::RedisClient(const RedisConf &conf)
{
	redisContext *cont = redisConnectWithTimeout(conf.host.c_str(), conf.port, conf.timeout);
	if (cont == nullptr)
		return;
	if (cont->err != 0)
	{
		redisFree(cont);
		return;
	}

	if (conf.pw.empty())
	{

	}
	else
	{
		std::string rediscmd = "AUTH " + conf.pw;
		redisReply *reply = reinterpret_cast<redisReply *>(redisCommand(cont, rediscmd.c_str()));
		if (!reply)
		{
			redisFree(cont);
			cont = nullptr;
			return ;
		}

		if (reply->type == REDIS_REPLY_ERROR) {
			redisFree(cont);
			freeReplyObject(reply);
			cont = nullptr;
			return ;
		}
	}

	context_ = cont;
}

bool as::db::RedisClient::Set(const std::string &keys, const std::string &val, const std::function<void(RedisResult*)> &fun)
{
	return RedisCommand(MakeSetCommand(keys, val), fun);
}

bool as::db::RedisClient::HashSet(const std::string &keys, const std::string &field, const std::string &val, const std::function<void(RedisResult *)> &fun)
{
	return RedisCommand(MakeHSetCommand(keys, field, val), fun);
}

bool as::db::RedisClient::Get(const std::string &keys, const std::function<void(as::db::RedisResult *)> &fun)
{
	return RedisCommand(MakeGetCommand(keys), fun);
}

bool as::db::RedisClient::HashGet(const std::string &keys, const std::string &field, const std::function<void(RedisResult *)> &fun)
{
	return RedisCommand(MakeHGetCommand(keys, field), fun);
}

bool as::db::RedisClient::HashGetAll(const std::string &keys, const std::function<void(RedisResult *)> &fun)
{
	return RedisCommand(MakeGetAllCommand(keys), fun);
}

void as::db::RedisClient::AsyncSet(const std::string &keys, const std::string &val, const std::function<void(RedisResult *)> &fun)
{
	std::string command = MakeSetCommand(keys, val);
	auto task = MakeTask(command, fun);
	list_lock_.lock();
	async_task_.push_back(task);
	list_lock_.unlock();
	list_co_.notify_one();
}

std::string as::db::RedisClient::MakeSetCommand(const std::string &keys, const std::string &val)
{
	return "SET " + keys + " " + val;
}

std::string as::db::RedisClient::MakeHSetCommand(const std::string &keys, const std::string &field, const std::string &val)
{
	return "HSET " + keys + " " + field + " " + val;
}

std::string as::db::RedisClient::MakeGetCommand(const std::string &keys)
{
	return "GET " + keys;
}

std::string as::db::RedisClient::MakeHGetCommand(const std::string &keys, const std::string &field)
{
	return "HGET " + keys + " " + field;
}

std::string as::db::RedisClient::MakeGetAllCommand(const std::string &keys)
{
	return "HGETALL " + keys;
}

void as::db::RedisClient::Tick()
{
	HandleTask();
}

as::db::RedisTask *as::db::RedisClient::MakeTask(const std::string &command,const std::function<void(RedisResult*)> &fun)
{
	auto time_now = std::chrono::system_clock::now();
	auto duration_in_ms = std::chrono::duration_cast<std::chrono::milliseconds>(time_now.time_since_epoch());

	auto task = new RedisTask();
	task->command = command;
	task->start_tick = duration_in_ms.count();
	task->cb = fun;
	return task;
}

bool as::db::RedisClient::HandleTask()
{
	std::unique_lock<std::mutex> l(list_lock_);
	list_co_.wait_for(l, std::chrono::seconds(ECON_WAIT_TIME), 
		[&]()
		{
			return async_task_.size() > 0;
		}
	);

	DoTask();
	return true;
}

void as::db::RedisClient::DoTask()
{
	while (async_task_.size() > 0)
	{
		bool hasError = false;
		for (auto it = async_task_.begin(); it != async_task_.end(); )
		{
			bool result = RedisCommand((*it)->command, (*it)->cb);
			if (result)
			{
				delete *it;
				it = async_task_.erase(it);
			}
			else
			{
				++it;
				hasError = true;
			}
		}

		if (hasError)
		{
			break;
		}
	}
}

bool as::db::RedisClient::RedisCommand(const std::string &keys, const std::function<void(RedisResult *)> &fun)
{
	redisReply *reply = (redisReply*)redisCommand(context_, keys.c_str());
	if (reply == nullptr)
	{
		switch (context_->err)
		{
		case REDIS_ERR_PROTOCOL:
		case REDIS_ERR_OOM:
		case REDIS_ERR_OTHER:
		{
			RedisResult result(nullptr);
			result.SetError(context_->errstr);
			result.SetErrorCode(RedisError::REDIS_FAILURE);
			fun(&result);
			break;
		}
		case REDIS_ERR_IO:
		case REDIS_ERR_EOF:
		case REDIS_ERR_TIMEOUT:
		{
			RedisResult result(nullptr);
			result.SetError(context_->errstr);
			result.SetErrorCode(RedisError::REDIS_TIMEOUT);
			fun(&result);
			break;
		}
		default:
			break;
		}
		reconnect_ = true;
		return false;
	}
	else
	{
		RedisResult result(reply);
		fun(&result);
		freeReplyObject(reply);
		return true;
	}
}

as::db::RedisClient::~RedisClient()
{
	list_co_.notify_all();
	{
		std::unique_lock<std::mutex> l(list_lock_);
		DoTask();
	}
	if (context_)
		redisFree(context_);
}

redisContext *as::db::RedisClient::GetContext()
{
	return context_;
}




#endif